/*  File Name	: EpollReactor.cpp
 *  Author		: Pang Jiacai
 *	 E-mail		: 823960168@qq.com
 *  Created on	: 2012-3-22
 *  Description:
 */
#include "EpollReactor.h"
#include "TimerHeap.h"
#include "Reactor.h"
#include "Timer.h"

EpollReactor::EventTuple::EventTuple(EventHandler *eh,ReactorMask mask)
{
	this->event_handler_ = eh;
	this->mask_ = mask;
	this->is_suspended_ = false;
	this->is_register_ = false;
}

EpollReactor::HandlerRepository::HandlerRepository()
{
	this->handler_ = NULL;
	this->cur_size_ =0;
	this->max_size_ =0;
}

EpollReactor::HandlerRepository::~HandlerRepository()
{
    
}

int EpollReactor::HandlerRepository::open(int max_size)
{
	this->max_size_ = max_size;
	if(this->max_size_ <=0)
	{
		return -1;
	}
	this->handler_ = new EventTuple[this->max_size_];
	if(NULL==this->handler_)
	{
		return -1;
	}
	return 0;
}

int EpollReactor::HandlerRepository::close(void)
{
	LockGuardReturn(RT_MUTEX,lock,this->lock_,-1);

	if(NULL!=this->handler_)
	{
		this->unbind_all();
		delete [] this->handler_;
		this->handler_ = NULL;
	}
	return 0;
}

int EpollReactor::HandlerRepository::bind(int handle,
		EventHandler *eh,ReactorMask mask)
{
	if( !this->is_legal_handler(handle) || NULL==eh)
	{
		return -1;
	}

	LockGuardReturn(RT_MUTEX,lock,this->lock_,-1);

	this->handler_[handle].event_handler_ = eh;
	this->handler_[handle].mask_ = mask;

	++this->cur_size_;
	return 0;
}

int EpollReactor::HandlerRepository::unbind(int handle)
{
	EventTuple *event = this->find(handle);
	if(NULL==event)
	{
		return -1;
	}

	LockGuardReturn(RT_MUTEX,lock,this->lock_,-1);
	event->event_handler_ =NULL;
	event->mask_ = EventHandler::NULL_MASK;
	--this->cur_size_;
	return 0;
}

int EpollReactor::HandlerRepository::unbind_all(void)
{
	LockGuardReturn(RT_MUTEX,lock,this->lock_,-1);

	for(int handle=0;handle<this->max_size_;handle++)
	{
		EventTuple *tuple = &this->handler_[handle];//this->find(handle);
		if(NULL==tuple)
		{
			continue;
		}

		if(NULL!=tuple->event_handler_)
		{
			tuple->event_handler_->handle_close(handle,tuple->mask_);
		}

		tuple->event_handler_ =NULL;
		tuple->mask_ = EventHandler::NULL_MASK;
		--this->cur_size_;
	}
	return 0;
}

bool EpollReactor::HandlerRepository::is_legal_handler(int handle)
{
	if(handle >=0 && handle < this->max_size_)
	{
		return true;
	}
	return false;
}

EpollReactor::EventTuple *EpollReactor::HandlerRepository::find(int handle)
{
	if(handle<0 || handle >this->max_size_)
	{
		return NULL;
	}

	LockGuardReturn(RT_MUTEX,lock,this->lock_,NULL);

	EventTuple *tmp= &this->handler_[handle];
	if(NULL==tmp->event_handler_)
	{
		return NULL;
	}
	return tmp;
}

int EpollReactor::HandlerRepository::cur_size(void)
{
	return this->cur_size_;
}

int EpollReactor::HandlerRepository::max_size(void)
{
	return this->max_size_;
}

//////////////////////////////////////////////////////////////////////////////////

EpollReactor::EpollReactor(int handle_size,int timer_heap_size,bool only_timer,SignalHanlder *sh)
	:init_(false),epoll_fd_(INVALID_HANDLER),sig_handler_(NULL),
	 del_sig_handler_(false),run_status_(0)
{
	this->open(handle_size,timer_heap_size,sh,only_timer);
}

EpollReactor::~EpollReactor()
{

}

int EpollReactor::open(int handle_size,int timer_heap_size,SignalHanlder *sh,bool only_timer)
{
	if(this->init_)
	{
		return -1;
	}

	this->only_timer_ = only_timer;

	int result = 0;
    
	//epoll init
	MEMZERO(&this->epoll_event_,sizeof(this->epoll_event_));

	this->epoll_fd_ = epoll_create(65535);
	if(epoll_fd_==INVALID_HANDLER)
	{
		result = -1;
	}
	this->epoll_event_.data.fd = INVALID_HANDLER;

	//handler init
	handle_size = handle_size > 0 ? handle_size : 2048;
	if (this->handler_rep_.open(handle_size) == -1)
	{
		result = -1;
	}

	if(!only_timer)
	{
		//signal handler
		this->sig_handler_ = sh;
		if(NULL==this->sig_handler_)
		{
			this->sig_handler_  = new SignalHanlder();
			del_sig_handler_ = true;
		}
		this->sig_handler_->open(this);
	}

	if(timer_heap_size>0)
	{
	    this->timer_heap_ = new TimerHeap(this);
	    if(NULL==this->timer_heap_)
	    {
	    	this->close();
	        return -1;
	    }
	    timer_heap_size = (timer_heap_size > 0) ? timer_heap_size : 1024;

	    if(this->timer_heap_->open(timer_heap_size)!=0)
	    {
	    	this->close();
	    	return -1;
	    }
	}

	if(result!=-1)
	{
		this->init_ = true;
	}
	else
	{
		LOG((DEBUG_LOG,"EpollReactor::open error"));
		this->close();
	}
	return result;
}

int EpollReactor::close(void)
{
	int result = 0;
	if(this->del_sig_handler_ && NULL !=this->sig_handler_)
	{
		delete this->sig_handler_;
		this->sig_handler_ = NULL;
	}

	if(NULL!=this->timer_heap_)
	{
		delete this->timer_heap_;
		this->timer_heap_ = NULL;
	}

	if(this->epoll_fd_ !=INVALID_HANDLER)
	{
		result = ::close(this->epoll_fd_);
	}

	this->handler_rep_.close();

	this->epoll_fd_ = INVALID_HANDLER;
	this->init_ = false;
	return result;
}

void EpollReactor::run_status(int status)
{
	if(status>=0 && status<=2)
	{
		this->run_status_ = status;
	}
}

//reactor run status,0:stop 1:start
int EpollReactor::get_run_status(void)
{
	return this->run_status_;
}

int EpollReactor::handle_events(int timeout)
{
	if (this->run_status_ == 0)
	{
		return -1;
	}
	int result = 0;

	if (this->epoll_event_.data.fd != INVALID_HANDLER)
	{
		result = 1;
	}
	else
	{
		for (HandleMap::iterator iter = this->resume_handle_map_.begin();
				iter != this->resume_handle_map_.end(); iter++)
		{
			EventTuple *tuple = this->handler_rep_.find(iter->first);
			if (NULL != tuple && tuple->event_handler_ == iter->second)
			{
				this->resume_handler(iter->first);
			}
		}

		this->resume_handle_map_.clear();

		int old_timeout = timeout;

		if(this->timer_heap_->timer_count()>0)
		{
			timeout = this->timer_heap_->next_timeout_value();
		}
		//if(this->only_timer_ && timeout==-1)
		if(timeout<=0)
		{
			timeout=0;
		}

//		if(this->only_timer_)
//		{
//			LOG((DEBUG_LOG,"EpollReactor::handle_events-[%d][%d][%d]",
//					this->only_timer_,old_timeout,timeout));
//		}
		result = ::epoll_wait(this->epoll_fd_, &this->epoll_event_, 1,timeout);
	}

	if ((result == -1 && errno == ETIME))
	{
		return 0;
	}
	else if (result == -1 && errno != EINTR)
	{
		return -1;
	}
	return this->dispatch_event();
}

int EpollReactor::dispatch_event(void)
{
	int result = 0;

	result = this->dispath_signal_event();
	if(result!=0)
	{
		return result;
	}

	result = this->dispatch_timer_event();
	if(result!=0)
	{
		return result;
	}

	result = this->dispatch_io_event();
	return result;
}

int EpollReactor::dispath_signal_event(void)
{
	if(this->epoll_event_.data.fd == this->sig_handler_->notify_fd())
	{
		const int handle = this->epoll_event_.data.fd;

		this->epoll_event_.data.fd = INVALID_HANDLER;
		this->epoll_event_.events = 0;

		int signum[1]={0};
		int recv_len = ::read(handle,signum,sizeof(int)*1);
		if(recv_len< (int)sizeof(int))
		{
			return 1;
		}
		this->sig_handler_->dispatch_signal_handler(signum[0]);

		this->resume_handler(handle);
		return 1;
	}
	return 0;
}

int EpollReactor::dispatch_timer_event(void)
{
	this->timer_heap_->handle_event();

	if(this->epoll_event_.data.fd==this->timer_heap_->timer_notify_fd[0])
	{
		int fd = this->epoll_event_.data.fd;
		this->epoll_event_.data.fd = INVALID_HANDLER;
		this->epoll_event_.events = 0;

		char buf[2]={0};
		if(::read(fd,buf,sizeof(char)*2)>0)
		{
			this->resume_handler(fd);
		}
		else
		{
			LOG((ERROR_LOG,"EpollReactor::dispatch_timer_event-->close_notify_fd"));
			this->timer_heap_->close_notify_fd();
		}
		return 1;
	}
	return 0;
}

int EpollReactor::dispatch_io_event(void)
{
	const uint32_t out_event = EPOLLOUT;
	const uint32_t exc_event = EPOLLPRI;
	const uint32_t in_event = EPOLLIN;
	const uint32_t err_event = EPOLLHUP | EPOLLERR;
	const int handle = this->epoll_event_.data.fd;
	uint32_t event = this->epoll_event_.events;
	this->epoll_event_.data.fd=INVALID_HANDLER;
	this->epoll_event_.events = 0;
	if (handle != INVALID_HANDLER)
	{
		EventTuple *tuple = this->handler_rep_.find(handle);
		if (NULL == tuple)
		{
			this->epoll_event_.data.fd = INVALID_HANDLER;
			return 0;
		}

		ReactorMask mask = 0;
		EventHandler *handler = tuple->event_handler_;
		int (EventHandler::*callback)(int handle);

		if (BIT_ENABLE(event,out_event))
		{
			mask = EventHandler::WRITE_MASK;
			callback = &EventHandler::handle_output;
		}
		else if (BIT_ENABLE(event,exc_event))
		{
			mask = EventHandler::EXCEPT_MASK;
			callback = &EventHandler::handle_exception;
		}
		else if (BIT_ENABLE(event,in_event))
		{
			mask = EventHandler::READ_MASK;
			callback = &EventHandler::handle_input;
		}
		else if (BIT_ENABLE(event,err_event))
		{
			this->remove_handler(handle, handler, tuple->mask_);
			return 1;
		}
		else
		{
			LOG((ERROR_LOG,"EpollReactor::dispatch_io_event unknow event"));
			return -1;
		}
		tuple->is_suspended_ = true;
		int result = 0;

		do
		{
			result=(handler->*callback)(handle);
		}
		while(result>0);

		if(result==0)
		{
			if(tuple->is_suspended_)
			{
				this->resume_handle_map_.insert(std::pair<int,
						EventHandler*>(handle,tuple->event_handler_));
			}
			return 1;
		}

		tuple = this->handler_rep_.find(handle);
		if(result <0 && NULL!=tuple && tuple->event_handler_==handler)
		{
			this->remove_handler(handle,tuple->event_handler_,tuple->mask_);
		}
		return 1;
	}
	return 0;
}

int EpollReactor::register_handler(int handle,EventHandler *eh,ReactorMask mask)
{
	LockGuardReturn(RT_MUTEX,lock,this->lock(),-1);

	if(handle==INVALID_HANDLER || mask==EventHandler::NULL_MASK)
	{
		return -1;
	}

	if(this->handler_rep_.find(handle)==NULL)
	{
		if(this->handler_rep_.bind(handle,eh,mask)!=0)
		{
			return -1;
		}

		EventTuple *tuple = this->handler_rep_.find(handle);

		const int op = EPOLL_CTL_ADD;
		struct epoll_event epev;
		MEMZERO(&epev,sizeof(struct epoll_event));
		epev.data.fd = handle;
		epev.events = this->mask_to_epool_event(mask);
		epev.events |= EPOLLONESHOT;

		if(::epoll_ctl(this->epoll_fd_,op,handle,&epev)==-1)
		{
			this->handler_rep_.unbind(handle);
			return -1;
		}
		tuple->is_register_ = true;
	}
	else
	{
		return this->mask_ops(handle,mask,Reactor::ADD_MASK);
	}
	LOG((DEBUG_LOG,"EpollReactor::register_handler:[%d]",handle));
	return 0;
}

int EpollReactor::resume_handler(int handle)
{
	EventTuple *tuple = this->handler_rep_.find(handle);
	if(NULL==tuple)
	{
		return -1;
	}
	if(tuple->mask_ == EventHandler::NULL_MASK)
	{
		return -1;
	}

	int op = EPOLL_CTL_ADD;
	if(tuple->is_register_)
	{
		op= EPOLL_CTL_MOD;
	}
	struct epoll_event epev;
	MEMZERO(&epev, sizeof(struct epoll_event));
	epev.data.fd = handle;
	epev.events = this->mask_to_epool_event(tuple->mask_);
	epev.events |= EPOLLONESHOT;

	//LOG((DEBUG_LOG,"EpollReactor::resume handle:[fd:%d][op:%d]",handle,op));

	if (::epoll_ctl(this->epoll_fd_, op, handle, &epev) == -1)
	{
		this->handler_rep_.unbind(handle);
		LOG((DEBUG_LOG,"EpollReactor::resume handle failed:[fd:%d][op:%d]",handle,op));
		return -1;
	}
	tuple->is_register_ = true;
	tuple->is_suspended_ = false;
	return 0;
}

/**
 * this function will call EventHandler::handle_close
 * so,don't invoke it in EventHandler::handle_close
 */
int EpollReactor::remove_handler(int handle,EventHandler *eh,ReactorMask mask)
{
	LockGuardReturn(RT_MUTEX,lock,this->lock(),-1);

	LOG((DEBUG_LOG,"EpollReactor::remove handle:[%d]",handle));

	EventTuple *tuple = this->handler_rep_.find(handle);
	if(NULL==tuple && NULL==eh)
	{
		LOG((DEBUG_LOG,"EpollReactor::remove handle failed 1:[%d]",handle));
		return -1;
	}

	bool register_change = true;
	if(NULL!=tuple && (NULL==eh || tuple->event_handler_==eh))
	{
		if(this->mask_ops(handle,mask,Reactor::CLR_MASK)<0)
		{
			LOG((DEBUG_LOG,"EpollReactor::remove handle failed 2:[%d]",handle));
			return -1;
		}
		register_change = false;
		eh = tuple->event_handler_;
	}

	if(NULL!=eh )//&& BIT_ENABLE(tuple->mask_,EventHandler::DONT_CALL))
	{
		eh->handle_close(handle,mask);
	}
	if(!register_change && tuple->mask_==EventHandler::NULL_MASK)
	{
		this->handler_rep_.unbind(handle);
	}
	return 0;
}

int EpollReactor::mask_ops(int handle,ReactorMask mask,int ops)
{
	EventTuple *tuple = this->handler_rep_.find(handle);
	if(NULL==tuple)
	{
		return -1;
	}

	const int old_mask = tuple->mask_;
	int new_mask = old_mask;

	switch(ops)
	{
		case Reactor::GET_MASK:
		{
			return old_mask;
		}
		case Reactor::ADD_MASK:
		{
			SET_BITS(new_mask,mask);
			SET_BITS(new_mask,mask);
			break;
		}
		case Reactor::SET_MASK:
		{
			new_mask = mask;
			break;
		}
		case Reactor::CLR_MASK:
		{
			CLR_BITS(new_mask,mask);
			break;
		}
		default:
			return -1;
	}

	tuple->mask_ = new_mask;

	if(!tuple->is_suspended_ ||
			(tuple->is_register_ && new_mask==EventHandler::NULL_MASK))
	{
		short const event = this->mask_to_epool_event(tuple->mask_);

		epoll_event ep;
		MEMZERO(&ep,sizeof(epoll_event));
		ep.data.fd = handle;
		ep.events = (event|EPOLLONESHOT);
		int op = 0;
		if(new_mask==EventHandler::NULL_MASK)
		{
			ep.events = 0;
			op = EPOLL_CTL_DEL;
		}
		else
		{
			op = EPOLL_CTL_MOD;
		}

		if(::epoll_ctl(this->epoll_fd_,op,handle,&ep)==-1)
		{
			/* If a handle is closed, epoll removes it from the poll set
			 * automatically - we may not know about it yet. If that's the
			 * case, a mod operation will fail with ENOENT. Retry it as an add.
			 */
			if ((op == EPOLL_CTL_MOD) && (errno == ENOENT))
			{
				if(::epoll_ctl(this->epoll_fd_,EPOLL_CTL_ADD,handle,&ep)==-1)
				{
					return -1;
				}
			}
		}
		tuple->is_register_ = (op!=EPOLL_CTL_DEL);
	}
	return old_mask;
}

short EpollReactor::mask_to_epool_event(ReactorMask mask)
{
	short event = 0;
	if(BIT_ENABLE(mask, EventHandler::READ_MASK)
			|| BIT_ENABLE(mask, EventHandler::CONNECT_MASK)
			|| BIT_ENABLE(mask, EventHandler::ACCEPT_MASK))
	{
		SET_BITS(event,EPOLLIN);
	}

	if(BIT_ENABLE(mask, EventHandler::WRITE_MASK)
			|| BIT_ENABLE(mask, EventHandler::CONNECT_MASK))
	{
		SET_BITS(event,EPOLLOUT);
	}

	if(BIT_ENABLE(mask, EventHandler::EXCEPT_MASK))
	{
		SET_BITS(event,EPOLLPRI);
	}

	return event;
}

int EpollReactor::register_signal(int signum,EventHandler *eh,
		bool replace_old,EventHandler *old_sh)
{
	return this->sig_handler_->register_signal(signum,eh,replace_old,old_sh);
}

int EpollReactor::remove_signal(int signal,EventHandler *eh)
{
	return this->sig_handler_->remove_signal(signal,eh);
}

long EpollReactor::schudle_timer(EventHandler *eh,TimeValue delay,TimeValue interval)
{
    return this->timer_heap_->register_timer(eh,delay,interval);
}

int EpollReactor::suspend_timer(long timer_id)
{
    return this->timer_heap_->suspend_timer(timer_id);
}

int EpollReactor::resume_timer(long timer_id)
{
    return this->timer_heap_->resume_timer(timer_id);
}

int EpollReactor::cancel_timer(long timer_id)
{
    return this->timer_heap_->remove_timer(timer_id);
}

int EpollReactor::refresh_timer(long timer_id,TimeValue new_time)
{
	return this->timer_heap_->refresh_timer(timer_id,new_time);
}

int EpollReactor::reactor_type(void)
{
	return 2;
}

RT_MUTEX &EpollReactor::lock(void)
{
	return this->reactor_lock_;
}
